package com.huateng.util;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.Map;

/**
 * Created by joy on 2016/11/4.
 */
@Component
public class HbaseUtils {
    private static Logger logger = Logger.getLogger(HbaseUtils.class);
    public static final String ROWKEY = "rowkey";
    private static Connection connection = null;
    @Resource
    private SecureUtils secureUtils;

    @PostConstruct
    public void init(){
        Configuration conf= secureUtils.getConf();
        try {
            connection = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(e);
        }
    }


    /**
     * 分页的结果不包含startrowkey的数据，
     * @param tablename
     * @param startrowkey
     * @param limit
     * @return
     */
    public List<Map<String,String>> querySimple(String tablename, String startrowkey, int limit,Filter filter){
        Table table = null;
        List<Map<String,String>> lists = Lists.newArrayList();
        try {
            table =  connection.getTable(TableName.valueOf(tablename));
            Scan scan = new Scan();
            scan.setCaching(1000);
            scan.setCacheBlocks(false);
            if(!Strings.isNullOrEmpty(startrowkey)){
                scan.setStartRow(Bytes.toBytes(startrowkey));
                limit = limit+1;
            }
            if(filter!=null){
                scan.setFilter(filter);
            }
            ResultScanner resultScanner = table.getScanner(scan);
            for(Result rs: resultScanner.next(limit)){
                Map<String,String> item = Maps.newHashMap();
                item.put(ROWKEY,new String(rs.getRow()));
                for(Cell cell:rs.rawCells()){   // 只关注列，不关注列簇
                    item.put(new String(CellUtil.cloneQualifier(cell)),new String(CellUtil.cloneValue(cell)));
                }
                lists.add(item);
            }
            //删除startrowkey对应的数据
            if(!Strings.isNullOrEmpty(startrowkey) && lists.size()>0){
                lists.remove(0);
            }
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(e);
        }finally {
            CloseUitl.close(table);
        }
        return lists;
    }


    /**
     * Filter策略
     *
     * mainstring.hashcode的前3位+mainstring, 后面是下限和上限
     * @return
     */
    public Filter generateFilter(String mainstr,String low,String top){
        FilterList filterlist  = null;
        if(!Strings.isNullOrEmpty(mainstr)){
            String startprefix=String.valueOf(mainstr.hashCode()).substring(0,3)+mainstr+low;
            Filter topfilter = new RowFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL
                    ,new BinaryComparator(startprefix.getBytes()));
            String endprefix=String.valueOf(mainstr.hashCode()).substring(0,3)+mainstr+top;
            Filter lowfilter = new RowFilter(CompareFilter.CompareOp.LESS_OR_EQUAL
                    ,new BinaryComparator(startprefix.getBytes()));
            List<Filter> filters = Lists.newArrayList();
            filters.add(topfilter);
            filters.add(lowfilter);
            filterlist=new FilterList(FilterList.Operator.MUST_PASS_ALL,filters);
        }
        return filterlist;
    }


    public Filter generatePreFilter(String prefix){
        Filter filter  = null;
        if(!Strings.isNullOrEmpty(prefix)){
            String startprefix=String.valueOf(prefix.hashCode()).substring(0,3)+prefix;
            filter = new RowFilter(CompareFilter.CompareOp.EQUAL
                    ,new BinaryPrefixComparator(startprefix.getBytes()));
        }
        return filter;
    }
}
